
#include "fiber-env.h"

#include <assert.h>
#include <stdio.h>

#include "cooperative-mutex.h"
#include "fiber.h"
#include "fiber-context.h"
#include "interlocked.h"
#include "thread.h"

namespace dulymoo {

FiberEnv::FiberEnv() :
    quit_flag_(false),
    fibers_(),
    fibers_lock_(),
    fibers_all_quitted_(fibers_lock_),
    ready_queue_(),
    sleeping_fibers_(),
    sleeping_fibers_lock_(),
    threads_(),
    poll_sleep_thread_(NULL),
    poll_sleep_thread_init_lock_()
{
}

FiberEnv::~FiberEnv()
{
}

bool FiberEnv::Initialize()
{
    return true;
}

void FiberEnv::Finalize()
{
}

bool FiberEnv::Start(unsigned int thread_count)
{
    const unsigned int kDefaultThreadCount = 8;
    const unsigned long kThreadStackSize = 256 * 1024;

    if (thread_count == 0)
        thread_count = kDefaultThreadCount;

    for (unsigned int i=0; i<thread_count; ++i)
        threads_.push_back(new Thread(ThreadProc, static_cast<void*>(this), kThreadStackSize));
    for (unsigned int i=0; i<thread_count; ++i)
        threads_[i]->Start();

    return true;
}

void FiberEnv::Stop()
{
    WaitAllFibersToQuit();

    quit_flag_ = true;

    const unsigned int thread_count = threads_.size();
    for (unsigned int i=0; i<thread_count; ++i)
        threads_[i]->Join();
    for (unsigned int i=0; i<thread_count; ++i)
        delete threads_[i];
    threads_.clear();

    if (poll_sleep_thread_ != NULL)
    {
        poll_sleep_thread_->Join();
        delete poll_sleep_thread_;
        poll_sleep_thread_ = NULL;
    }
}

bool FiberEnv::AddFiber(void (*func)(void*), void* arg, const char* name)
{
    FiberContext* fiber_context = new FiberContext(func, arg, 0, name);
    fiber_context->SetState("ready");

    fibers_lock_.Lock();
    fibers_[fiber_context->id_] = fiber_context;
    fibers_lock_.Unlock();

    return ready_queue_.Push(fiber_context);
}

void FiberEnv::DumpFibers(FILE *fp)
{
    fibers_lock_.Lock();
    for (fibers_t::iterator iter = fibers_.begin();
        iter != fibers_.end(); ++iter)
    {
        FiberContext *fiber_context = iter->second;
        fprintf(fp, "fiber #%d, name = %s, state = %s\n",
            fiber_context->id_, fiber_context->name_.c_str(), fiber_context->state_);
    }
    fibers_lock_.Unlock();
}

/////////////////////////////////////////////////////////////////////////////

ISocket* FiberEnv::NewSocket(...)
{
    // TODO:
    return NULL;
}

IFile* FiberEnv::NewFile(...)
{
    // TODO:
    return NULL;
}

IMutex* FiberEnv::NewMutex(bool recursive)
{
    assert (GetThreadContext() != NULL);
    return new CooperativeMutex(recursive);
}

IRWLock* FiberEnv::NewRWLock(...)
{
    // TODO:
    return NULL;
}

ICondition* FiberEnv::NewCondition(...)
{
    // TODO:
    return NULL;
}

IEvent* FiberEnv::NewEvent(...)
{
    // TODO:
    return NULL;
}

ISemaphore* FiberEnv::NewSemaphore(...)
{
    // TODO:
    return NULL;
}

void FiberEnv::Sleep(unsigned long milliseconds)
{
    if (milliseconds == 0)
    {
        Yield();
        return;
    }

    ThreadContext *thread_context = GetThreadContext();
    assert (thread_context != NULL);
    thread_context->Schedule(CompleteSleep, reinterpret_cast<void*>(milliseconds));
}

void FiberEnv::CompleteSleep(ThreadContext* thread_context, void* arg)
{
    unsigned long milliseconds = reinterpret_cast<unsigned long>(arg);

    ABSTime time_to_wake;
    time_to_wake.Adjust(0, milliseconds*1000);

    thread_context->current_fiber_->SetState("sleep");

    FiberEnv *env = thread_context->env_;
    env->sleeping_fibers_lock_.Lock();
    env->sleeping_fibers_.insert(std::make_pair(time_to_wake, thread_context->current_fiber_));
    env->sleeping_fibers_lock_.Unlock();

    if (env->poll_sleep_thread_ == NULL)
        env->CreatePollSleepThread();
}

void FiberEnv::Yield()
{
    ThreadContext *thread_context = GetThreadContext();
    assert (thread_context != NULL);

    FiberEnv *env = thread_context->env_;
    unsigned int ready_queue_size = env->ready_queue_.Size();
    if (ready_queue_size == 0)
        // no other ready fiber, continue current one
        return;

    thread_context->Schedule(CompleteYield, NULL);
}

void FiberEnv::CompleteYield(ThreadContext* thread_context, void*)
{
    thread_context->current_fiber_->SetState("ready");
    thread_context->env_->ready_queue_.Push(thread_context->current_fiber_);
}

void FiberEnv::Quit()
{
    ThreadContext *thread_context = GetThreadContext();
    assert (thread_context != NULL);
    thread_context->Schedule(CompleteQuit, NULL);
}

void FiberEnv::CompleteQuit(ThreadContext* thread_context, void*)
{
    FiberEnv *env = thread_context->env_;

    FiberContext *fiber_context = thread_context->current_fiber_;
    fiber_context->SetState("quit");

    env->fibers_lock_.Lock();
    env->fibers_.erase(fiber_context->id_);
    if (env->fibers_.empty())
        env->fibers_all_quitted_.Broadcast();
    env->fibers_lock_.Unlock();

    delete fiber_context;
}

void FiberEnv::SetCurrentFiberState(const char* state)
{
    ThreadContext* thread_context = GetThreadContext();
    assert (thread_context != NULL);
    thread_context->current_fiber_->SetState(state);
}

bool FiberEnv::CreateFiber(void (*func)(void*), void* arg, const char* name)
{
    ThreadContext* thread_context = GetThreadContext();
    assert (thread_context != NULL);
    FiberEnv* env = thread_context->env_;
    assert (env != NULL);

    return env->AddFiber(func, arg, name);
}

/////////////////////////////////////////////////////////////////////////////

FiberContext* FiberEnv::ThreadContext::GetCurrentFiber()
{
    return current_fiber_;
}

void FiberEnv::ThreadContext::ReadyFiber(FiberContext* fiber_context)
{
    assert (fiber_context != NULL);
    fiber_context->SetState("ready");
    env_->ready_queue_.Push(fiber_context);
}

bool FiberEnv::ThreadContext::Schedule()
{
    post_switch_action_ = NULL;
    return schedule_fiber_->SwitchTo(current_fiber_->fiber_);
}

bool FiberEnv::ThreadContext::Schedule(void (*func)(ThreadContext*, void*), void*arg)
{
    post_switch_action_ = func;
    post_switch_arg_    = arg;
    return schedule_fiber_->SwitchTo(current_fiber_->fiber_);
}

FiberEnv::ThreadContext* FiberEnv::GetThreadContext()
{
    return reinterpret_cast<ThreadContext*>(Thread::GetSpecific());
}

/////////////////////////////////////////////////////////////////////////////

void FiberEnv::ThreadProc(void* arg)
{
    FiberEnv* env = reinterpret_cast<FiberEnv*>(arg);
    if (env != NULL)
        env->ThreadProc();
}

void FiberEnv::ThreadProc()
{
    Fiber this_fiber;

    ThreadContext thread_context;
    thread_context.env_                = this;
    thread_context.schedule_fiber_     = &this_fiber;
    thread_context.current_fiber_      = NULL;
    thread_context.post_switch_action_ = NULL;
    thread_context.post_switch_arg_    = NULL;

    Thread::SetSpecific(&thread_context);

    while (!quit_flag_)
    {
        FiberContext* fiber = ready_queue_.Pop(true, 10);
        if (fiber == NULL)
            continue;

        fiber->SetState("running");
        thread_context.current_fiber_ = fiber;
        fiber->fiber_.SwitchTo(this_fiber);
        if (thread_context.post_switch_action_ != NULL)
        {
            (thread_context.post_switch_action_)(&thread_context, thread_context.post_switch_arg_);
            thread_context.post_switch_action_ = NULL;
        }
    }
}

void FiberEnv::PollSleepThreadProc(void* arg)
{
    FiberEnv* env = reinterpret_cast<FiberEnv*>(arg);
    if (env != NULL)
        env->PollSleepThreadProc();
}

void FiberEnv::PollSleepThreadProc()
{
    while (!quit_flag_)
    {
        ABSTime now;

        sleeping_fibers_lock_.Lock();
        sleeping_fibers_t::iterator upper_bound = sleeping_fibers_.upper_bound(now);
        sleeping_fibers_t::iterator iter = sleeping_fibers_.begin();
        while (iter != upper_bound)
        {
            ready_queue_.Push(iter->second);
            sleeping_fibers_.erase(iter++);
        }
        sleeping_fibers_lock_.Unlock();

        usleep(1000*10);
    }
}

void FiberEnv::CreatePollSleepThread()
{
    const unsigned long kPollSleepThreadStackSize = 256*1024;

    poll_sleep_thread_init_lock_.Lock();
    if (poll_sleep_thread_ == NULL)
    {
        poll_sleep_thread_ = new Thread(PollSleepThreadProc,
                reinterpret_cast<void*>(this), kPollSleepThreadStackSize);
        poll_sleep_thread_->Start();
    }
    poll_sleep_thread_init_lock_.Unlock();
}

void FiberEnv::WaitAllFibersToQuit()
{
    fibers_lock_.Lock();
    while (!fibers_.empty())
        fibers_all_quitted_.Wait();
    fibers_lock_.Unlock();
}

} // namespace dulymoo

